package com.shujia.flink.transformation

import org.apache.flink.streaming.api.scala._

object Demo5Agg {

  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    val ds: DataStream[String] = env.readTextFile("flink/data/students.txt")

    /**
      * 取出每个班级年龄最大的学生
      *
      */

    ds.map(line => {
      val split: Array[String] = line.split(",")
      Student(split(0), split(1), split(2).toInt, split(3), split(4))
    })
      .keyBy(_.clazz)
      .maxBy("age")
      .print()


    env.execute()
  }

  case class Student(id: String, name: String, age: Int, gender: String, clazz: String)

}
